#include "NodeInternalServer.hpp"
#include "cybertron/network/SocketEventHandler.hpp"
#include "cybertron/network/Message.hpp"
#include "cybertron/core/UtilTime.hpp"
#include "cybertron/core/Log.hpp"
#include "Node/Timer.pb.h"

CYBERTRON_BEGIN

class EventHandlerNodeConnection : public SocketEventHandler
{
public:
	EventHandlerNodeConnection(std::shared_ptr<NodeInternalServer> pServer)
	{
		mServer = pServer;
	}

	virtual void onMessage(Message& msg);
	virtual void onDisconnected();

private:
	std::weak_ptr<NodeInternalServer> mServer;
};

void EventHandlerNodeConnection::onMessage(Message& msg)
{
	std::uint16_t msgId = msg.parseMsgId();

	if (msgId == Timer::EFromAnyDeclareRole)
	{
		Timer::FromAnyDeclareRole declare;
		if (!msg.toProtobuf(declare))
		{
			close("Invalid Protocol");
			return;
		}

		bool succeeded = false;

		std::shared_ptr<NodeInternalServer> pServer = mServer.lock();
		if (pServer != nullptr)
		{
			if (declare.role() == Timer::ERole_Console)
			{
				pServer->addConsole(getSocket());
				succeeded = true;
			}
		}

		Timer::FromAnyResultDeclareRole result;
		result.set_succeeded(succeeded);
		send(Timer::EFromAnyResultDeclareRole, result);
	}
	else if(msgId == Timer::EFromDaemonNodeAreYouAlive)
	{
		Message areYouAliveMsg;
		areYouAliveMsg.create(Timer::EFromDaemonNodeAreYouAlive, 0);
		send(areYouAliveMsg);
	}
	else
	{
		std::shared_ptr<NodeInternalServer> pServer = mServer.lock();
		if (pServer != nullptr)
		{
			pServer->queueMessage(getSocket(), msg);
		}
	}
}

void EventHandlerNodeConnection::onDisconnected()
{
	std::shared_ptr<NodeInternalServer> pServer = mServer.lock();
	if (pServer)
	{
		auto clientConnectionHandler = pServer->clientConnectionHandler();
		if(clientConnectionHandler){
			clientConnectionHandler(getSocket(), ESocketStatus_Closed);
		}
	}
}

NodeInternalServer::NodeInternalServer()
{
	// do nothing.
}

NodeInternalServer::~NodeInternalServer()
{
	// do nothing.
}

void NodeInternalServer::close()
{
	mHistoryLogs.clear();
	mMessageQueue.clear();

	{
		std::unique_lock<std::recursive_mutex> lock(mMutexConsoles);
		mConsoles.clear();
		mNewConsoles.clear();
	}

	SocketTcpServer::close();
}

void NodeInternalServer::onAccept(SocketBasePtr pRemoteSocket)
{
	if (pRemoteSocket == nullptr)
	{
		return;
	}

	std::shared_ptr<NodeInternalServer> self = 
		std::static_pointer_cast<NodeInternalServer>(shared_from_this());

	std::shared_ptr<EventHandlerNodeConnection> pHandler =
		std::make_shared<EventHandlerNodeConnection>(self);
	pRemoteSocket->addHandler(pHandler);

	// tell Node that some client is connected.
	if(mClientConnectionHandler){
		mClientConnectionHandler(pRemoteSocket, ESocketStatus_Connected);
	}
}

void NodeInternalServer::dispatchEventLog(
	std::uint32_t type,
	const std::string& text)
{
	Timer::FromNodeEventLog log;

	std::string strTime = UtilTime::getLocalTimeStringForPrint();
	log.set_time(strTime);
	log.set_type(type);
	log.set_text(text);

	Message msg;
	msg.fromProtobuf(Timer::EFromNodeEventLog, log);

	mHistoryLogs.push_back(msg);

	dispatchToConsole(msg);
}

void NodeInternalServer::dispatchToConsole(const Message& msg)
{
	mMainThreadTempArray.clear();
	getConsoles(mMainThreadTempArray);

	for (size_t i = 0; i < mMainThreadTempArray.size(); ++i)
	{
		Message msgClone = msg;
		mMainThreadTempArray[i]->send(msg);
	}

	mMainThreadTempArray.clear();
}

void NodeInternalServer::setMessageDirectHandler(
	std::function<bool(SocketBasePtr, const Message&)> handler)
{
	mMessageDirectHandler = handler;
}

void NodeInternalServer::setMessageHandler(
	std::function<void(SocketBasePtr, const Message&)> handler)
{
	mMessageHandler = handler;
}

void NodeInternalServer::setClientConnectionHandler(
	std::function<void(SocketBasePtr, ESocketStatus)> handler)
{
	mClientConnectionHandler = handler;
}

std::function<void(SocketBasePtr, ESocketStatus)> NodeInternalServer::clientConnectionHandler() const
{
	return mClientConnectionHandler;
}

void NodeInternalServer::queueMessage(SocketBasePtr pSocket, const Message& msg)
{
	if (pSocket == nullptr)
	{
		return;
	}

	if (mMessageDirectHandler != nullptr)
	{
		if (mMessageDirectHandler(pSocket, msg))
		{
			return;
		}
	}

	Command cmd;
	cmd.pSender = pSocket;
	cmd.msg = msg;

	mMessageQueue.push_back(cmd);
}

void NodeInternalServer::onTimer()
{
	// dispatch history logs to new console
	{
		mMainThreadTempArray.clear();
		{
			std::unique_lock<std::recursive_mutex> lock(mMutexConsoles);
			for (auto it = mNewConsoles.begin(); it != mNewConsoles.end(); ++it)
			{
				mMainThreadTempArray.push_back((*it));
				mConsoles.insert((*it));
			}
			mNewConsoles.clear();
		}

		for (size_t i = 0; i < mMainThreadTempArray.size(); ++i)
		{
			for (auto it = mHistoryLogs.begin(); it != mHistoryLogs.end(); ++it)
			{
				Message msgClone = (*it);
				mMainThreadTempArray[i]->send(msgClone);
			}
		}

		mMainThreadTempArray.clear();
	}

	if (mMessageHandler == nullptr)
	{
		mMessageQueue.clear();
	}
	else
	{
		Command cmd;
		while (mMessageQueue.pop_front(std::move(cmd)))
		{
			mMessageHandler(cmd.pSender, cmd.msg);
		}
	}
}

void NodeInternalServer::addConsole(SocketBasePtr pSocket)
{
	if (pSocket == nullptr)
	{
		return;
	}

	std::unique_lock<std::recursive_mutex> lock(mMutexConsoles);
	mNewConsoles.insert(pSocket);
}

void NodeInternalServer::delConsole(SocketBasePtr pSocket)
{
	if (pSocket == nullptr)
	{
		return;
	}

	std::unique_lock<std::recursive_mutex> lock(mMutexConsoles);
	auto it = mConsoles.find(pSocket);
	if (it != mConsoles.end())
	{
		mConsoles.erase(it);
	}
}

void NodeInternalServer::getConsoles(std::vector<SocketBasePtr>& sockets)
{
	std::unique_lock<std::recursive_mutex> lock(mMutexConsoles);
	for (auto it = mConsoles.begin(); it != mConsoles.end(); ++it)
	{
		sockets.push_back((*it));
	}
}

CYBERTRON_END
